package com.lvmama.rhino.analyze.push

import java.util.Date

import com.lvmama.rhino.common.entity.JobConfig
import com.lvmama.rhino.common.utils.JDBCUtil.JDBCTemplate
import com.lvmama.rhino.common.utils.spark.SparkApplication
import org.apache.commons.lang3.time.{DateFormatUtils, DateUtils}
import org.apache.spark.sql.{SQLContext, SaveMode}

/**
  * @Title: InvalidDeviceMark
  * @Package com.lvmama.rhino.analyze.push
  * @Description: 用于生成自定义用效设备记录，每天执行一次
  * @author 吴川通
  * @date 2016/11/15 17:53
  * @version V1.0
  */
class CustomInvalidDeviceCollect(config: JobConfig, days: Int) extends SparkApplication {

  override var appName: String = "CustomInvalidDeviceCollect"
  override var sparkConfig: Map[String, String] = config.spark

  val TBL_PUSH_DEVICEINFO: String = "push_deviceInfo"
  val TBL_PUSH_MI_REGID: String = "push_mi_regId"
  val TBL_PUSH_LOCATIONINFO: String = "push_locationInfo"
  val FAM_BASEINFO: String = "baseInfo"

  val JDBCDefault = JDBCTemplate.JDBCDefaultSet
  val connP = JDBCTemplate.getConProperties

  def execute(): Unit = {
    sparkConfig += ("spark.app.name" -> appName)

    val daysVal = days
    val curDate = new Date()
    val curDateStr = DateFormatUtils.format(curDate, "yyyy-MM-dd HH:mm:ss")
    val defaultCreateTime = "2016-11-01 00:00:00"
    val splitTime = DateUtils.parseDate("2017-02-15 00:00:00", "yyyy-MM-dd HH:mm:ss")

    withSparkContext { sc =>

      //1.自定义无效设备处理，查出来最后一次登录时间超过指定天数的，并且状态是空或有效的记录，标识成无效设备
      import com.lvmama.rhino.common.utils.hbase._
      val sqlCtx = new SQLContext(sc)
      import sqlCtx.implicits._
      //1.1 自定义无效设备，置 push_locationInfo 表为无效状态,
      //大于指定日期的无效记录插入MYSQL中的push_invalid_device表中,由hippo定时任务处理，小于指定日期情况直接在spark中处理，不在hippo中处理
      if ((curDate compareTo splitTime) > 0) {
        //插入MYSQL中
        val devicesRdd4Location = sc.hbaseTable[(Option[String], Option[String], Option[String], Option[String])](TBL_PUSH_DEVICEINFO)
          .select("lastTime", "valid", "objectId", "platform").inColumnFamily(FAM_BASEINFO).filter(f => {
          val lastTime = DateUtils.parseDate(f._1.getOrElse(defaultCreateTime), "yyyy-MM-dd HH:mm:ss")
          val compare = DateUtils.addDays(lastTime, daysVal) compareTo curDate
          compare <= 0 && (f._2.isEmpty || f._2.get == "" || f._2.get.equals("1"))
        }).map(m => (m._4.get, "3", m._3, "0", curDateStr, curDateStr)).toDF("platform", "type", "object_id", "status", "create_time", "modify_time")

        devicesRdd4Location.write.mode(SaveMode.Append).option("driver", JDBCDefault("driver"))
          .jdbc(JDBCTemplate.getValue("jdbc.hippo.conn"), "push_invalid_device", connP)

      } else {
        // 设备表  超过指定天数未登录 && (valid is null || valid == "1" )
        val devicesRdd4Location = sc.hbaseTable[(Option[String], Option[String], Option[String])](TBL_PUSH_DEVICEINFO)
          .select("lastTime", "valid", "objectId").inColumnFamily(FAM_BASEINFO).filter(f => {
          val lastTime = DateUtils.parseDate(f._1.getOrElse(defaultCreateTime), "yyyy-MM-dd HH:mm:ss")
          val compare = DateUtils.addDays(lastTime, daysVal) compareTo curDate
          compare <= 0 && (f._2.isEmpty || f._2.get == "" || f._2.get.equals("1"))
        }).map(m => m._3.get).toDF("objectId")
        //省份订阅表 push_locationInfo条件  miValid is null || miValid == "1" ||
        val locationInfoRdd = sc.hbaseTable[(String, Option[String], Option[String], Option[String])](TBL_PUSH_LOCATIONINFO)
          .select("objectId", "miValid", "gtValid").inColumnFamily(FAM_BASEINFO)
          .filter(f => f._3.isEmpty || f._3.get == "" || (f._3.get equals "1") || f._4.isEmpty || f._4.get == "" || (f._4.get equals "1"))
          .map(m => (m._1, m._2)).toDF("rowKey", "objectId")

        //更新 push_locationInfo 表中 miValid、gtValid 为无效状态
        locationInfoRdd.join(devicesRdd4Location, Seq("objectId")).
          map(m => (m.get(1).toString, "0", "0")).rdd.toHBaseTable(TBL_PUSH_LOCATIONINFO).toColumns("miValid", "gtValid").inColumnFamily(FAM_BASEINFO).save()

      }

      //1.1 自定义无效设备，置 push_deviceInfo 表记录为无效状态，无效判断条件为：超过指定天数未登录&&(valid is null || valid == "1" )
      val devicesRdd = sc.hbaseTable[(String, Option[String], Option[String], Option[String])](TBL_PUSH_DEVICEINFO)
        .select("lastTime", "valid", "objectId").inColumnFamily(FAM_BASEINFO).filter(f => {
        val lastTime = DateUtils.parseDate(f._2.getOrElse(defaultCreateTime), "yyyy-MM-dd HH:mm:ss")
        val compare = DateUtils.addDays(lastTime, daysVal) compareTo curDate
        compare <= 0 && (f._3.isEmpty || f._3.get == "" || f._3.get.equals("1"))
      })

      devicesRdd.map(m => (m._1, "0")).toHBaseTable(TBL_PUSH_DEVICEINFO).toColumns("valid").inColumnFamily(FAM_BASEINFO).save()

    }
  }

}

object CustomInvalidDeviceCollect {
  val config = JobConfig()

  def apply(days: Int): Unit = new CustomInvalidDeviceCollect(config, days).execute()
}
